package com.hxl.alert.consumer;

import com.hxl.alert.constant.CacheConstants;
import com.hxl.alert.entity.AlertOriginBase;
import com.hxl.alert.entity.FailedAlertMessage;
import com.hxl.alert.exception.AlertValidationException;
import com.hxl.alert.utils.AlertTimeUtils;
import com.hxl.alert.utils.RedisUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.text.ParseException;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Component
@RocketMQMessageListener(
        topic = "ALERT_TOPIC",
        consumerGroup = "ALERT_CONSUMER_GROUP1",
        consumeMode = ConsumeMode.CONCURRENTLY,
        messageModel = MessageModel.CLUSTERING
)
public class AlertConsumer implements RocketMQListener<AlertOriginBase> {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Resource
    private RedisUtils redisUtils;

    @Override
    public void onMessage(AlertOriginBase alert) {
        try {
            validateAlert(alert);
        } catch (AlertValidationException e) {
            e.printStackTrace();
            handleFailedAlert(alert, e.getMessage());
        }
        String faultId = alert.getFaultId();
        String lockKey = CacheConstants.PROCESS_LOCK + faultId;
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), alert);
        try {
            // 获取分布式锁
            Boolean locked = redisUtils.setIfAbsent(lockKey, "locked", 30, TimeUnit.SECONDS);

            if (Boolean.TRUE.equals(locked)) {
//                logger.info("获取锁成功");
                processAlert(alert);
            } else {
                // 稍后重试
                Thread.sleep(100);
                onMessage(alert);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            handleFailedAlert(alert, "系统错误: " + e.getMessage());
        } finally {
            redisUtils.delete(lockKey);
        }
    }

    private void processAlert(AlertOriginBase alert) {
//        logger.info("开始处理告警信息");
        try {
            //先将所有告警存入redis
            handleReceiveMessage(alert);
            if ("越限".equals(alert.getFaultDescr())) {
                handleYueXian(alert);
            } else {
                handleFault(alert);
            }
        } catch (ParseException e) {
            // 处理时间格式异常
            e.printStackTrace();
        }
    }

    private void handleReceiveMessage(AlertOriginBase alert) throws ParseException {
//        logger.info("开始处理接收到的告警信息");
        redisUtils.set(CacheConstants.RECEIVE_KEY + alert.getFaultId(), alert);
        redisUtils.zAdd(CacheConstants.SCORE_KEY, alert.getFaultId(), AlertTimeUtils.toTimestamp(alert.getOccurTime()));
    }

    private void handleYueXian(AlertOriginBase alert) throws ParseException {
//        logger.info("开始处理越限逻辑");
        // 存储越限告警
        redisUtils.set(CacheConstants.YUEXIAN_KEY + alert.getFaultId(), alert);

        // 设置检查时间点（越限时间 + 3分钟）
        long checkTime = AlertTimeUtils.toTimestamp(alert.getOccurTime()) + 180_000;
        redisUtils.zAdd(CacheConstants.PENDING_CHECKS, alert.getFaultId(), checkTime);
    }

    private void handleFault(AlertOriginBase alert) throws ParseException {
//        logger.info("开始处理普通故障逻辑");
        // 仅存储故障告警
        redisUtils.zAdd(CacheConstants.FAULT_KEY, alert.getFaultId(), AlertTimeUtils.toTimestamp(alert.getOccurTime()));
    }

    /**
     * 告警消息校验
     */
    private void validateAlert(AlertOriginBase alert) throws AlertValidationException {
        // 校验faultId
        if (StringUtils.isEmpty(alert.getFaultId())) {
            throw new AlertValidationException("faultId不能为空");
        }

        // 校验occurTime
        if (StringUtils.isEmpty(alert.getOccurTime())) {
            throw new AlertValidationException("occurTime不能为空");
        }

        try {
            AlertTimeUtils.toTimestamp(alert.getOccurTime());
        } catch (ParseException e) {
            throw new AlertValidationException("时间格式错误: " + alert.getOccurTime());
        }
    }

    /**
     * 处理失败消息
     */
    private void handleFailedAlert(AlertOriginBase alert, String reason) {
        FailedAlertMessage failedAlert = new FailedAlertMessage();
        failedAlert.setOriginalAlert(alert);
        failedAlert.setErrorReason(reason);
        failedAlert.setFailedTime(new Date());

        // 存储到Redis失败队列
        redisUtils.lRightPush(CacheConstants.FAILED_ALERTS_KEY, failedAlert);

        // 设置过期时间（7天）
        redisUtils.expire(CacheConstants.FAILED_ALERTS_KEY, 7, TimeUnit.DAYS);

        logger.error("消息处理失败: {}", reason);
    }
}